package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.TurnoverRateBean;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.mail.MailSend;
import cn.itcast.mail.MailUtils;
import cn.itcast.util.DbUtil;
import cn.itcast.util.RedisUtil;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.List;
import java.util.Map;

//换手率实时预警开发
public class TurnoverRateTask implements ProcessDataInterface {

    @Override
    public void process(DataStream<CleanBean> waterData) {
        //todo  1.创建bean对象
        //todo 2.数据转换process
        SingleOutputStreamOperator<TurnoverRateBean> processData = waterData.process(new ProcessFunction<CleanBean, TurnoverRateBean>() {
            //Map<sec_code,Map<字段名,字段值>>
            Map<String, Map<String, Object>> stockMap = null;

            @Override
            public void open(Configuration parameters) throws Exception {
                //todo   (1)加载mysql流通股本数据
                String sql = "select * from  bdp_sector_stock";
                stockMap = DbUtil.query("sec_code", sql); //获得每一个股数据
            }

            @Override
            public void processElement(CleanBean cleanBean, ProcessFunction<CleanBean, TurnoverRateBean>.Context context, Collector<TurnoverRateBean> collector) throws Exception {
                //todo   (2)封装bean对象数据
                Map<String, Object> map = stockMap.get(cleanBean.getSecCode());
                if(map!=null) {
                    Long nego_cap = Long.valueOf(map.get("nego_cap").toString());
                    TurnoverRateBean turnoverRateBean = new TurnoverRateBean();
                    turnoverRateBean.setSecCode(cleanBean.getSecCode());
                    turnoverRateBean.setSecName(cleanBean.getSecName());
                    turnoverRateBean.setTradePrice(cleanBean.getTradePrice());
                    turnoverRateBean.setTradeVol(cleanBean.getTradeVolumn());
                    turnoverRateBean.setNegoCap(nego_cap);
                    collector.collect(turnoverRateBean);
                }
            }
        });
        //todo 3.加载redis换手率数据
        JedisCluster jedisCluster = RedisUtil.getJedisCluster();
        //换手率阈值
        BigDecimal threshold = new BigDecimal(jedisCluster.hget("quot", "turnoverRate"));
        //todo 4.模式匹配
        Pattern<TurnoverRateBean, TurnoverRateBean> pattern = Pattern.<TurnoverRateBean>begin("begin")
                .where(new SimpleCondition<TurnoverRateBean>() {
                    @Override
                    public boolean filter(TurnoverRateBean value) throws Exception {
                        //换手率=成交量/流通股本×100%
                        BigDecimal turnoverRate = new BigDecimal(value.getTradeVol()).divide(new BigDecimal(value.getNegoCap()), 2, RoundingMode.HALF_UP);
                        if (turnoverRate.compareTo(threshold) == 1) {
                            return true;
                        }
                        return false; //正常数据
                    }
                });
        //todo 实时流匹配模式规则
        PatternStream<TurnoverRateBean> cep = CEP.pattern(processData.keyBy(TurnoverRateBean::getSecCode), pattern);
        cep.select(new PatternSelectFunction<TurnoverRateBean, Object>() {
            @Override
            public Object select(Map<String, List<TurnoverRateBean>> map) throws Exception {
                List<TurnoverRateBean> begin = map.get("begin");
                if(begin!=null && begin.size()>0){
                    MailSend.send(begin.toString());
                }
                return begin;
            }
        }).print();
        //todo 5.查询数据
        //todo 6.发送告警邮件
        //todo 7.数据打印
    }
}
